package edu.polsl.bio.str.manager.logic;

import edu.polsl.bio.str.common.constant.JobStatus;
import edu.polsl.bio.str.common.entity.Job;
import edu.polsl.bio.str.common.entity.JobGroup;
import edu.polsl.bio.str.common.entity.Node;
import edu.polsl.bio.str.common.exception.ManagerException;
import edu.polsl.bio.str.common.messaging.StrJobPayload;
import edu.polsl.bio.str.manager.dao.JobDao;
import edu.polsl.bio.str.manager.dao.JobDataDao;
import edu.polsl.bio.str.manager.dao.JobGroupDao;
import edu.polsl.bio.str.manager.dao.NodeDao;
import edu.polsl.bio.str.manager.error.NodeCommunicationErrorHandler;
import edu.polsl.bio.str.manager.messaging.ManagerMessagePublisher;
import edu.polsl.bio.str.manager.messaging.QueueLocator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.message.GenericMessage;
import org.springframework.transaction.annotation.Isolation;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import java.util.Date;
import java.util.List;

/**
 * //todo class description
 * <p/>
 * Created on: Jul 13, 2011 5:53:09 PM <br/>
 * <a href="http://www.polsl.pl">www.polsl.pl</a>
 *
 * @author lukasz.olczak@polsl.pl
 */
public class JobDispatcherBeanImpl implements JobDispatcherBean {
  /**
   * slf4j Logger.
   */
  private static final Logger log = LoggerFactory.getLogger(JobDispatcherBeanImpl.class.getName());

  private JobDao jobDao;

  private JobDataDao jobDataDao;

  private JobGroupDao jobGroupDao;

  private NodeDao nodeDao;

  private ManagerMessagePublisher messagePublisher;

  private QueueLocator queueLocator;

  private NodeCommunicationErrorHandler errorHandler;

  @Override
  @Transactional(propagation = Propagation.REQUIRED, isolation = Isolation.SERIALIZABLE)
  public void dispatchWaitingJobs() {
    List<Node> nodes = nodeDao.getIdleNodes();
    int sumOfTokens = sumOfTokens(nodes);
    if (sumOfTokens == 0) {
      log.info("There is no available tokens - nothing to do");
      return;
    }
    List<Job> jobs = jobDao.getWaitingJobs(sumOfTokens(nodes));
    if (jobs.size() == 0) {
      log.info("There is nothing to do, no waiting jobs = 0");
      return;
    }

    for (Job job : jobs) {
      Node node = nodes.get(0);

      sendJob(job, nodes.get(0));
      if (node.getTokens() == 0) {
        nodes.remove(node);
      }
    }
  }

  private void sendJob(Job job, Node node) {
    byte[] jobData = jobDataDao.getJobDataAsByteArray(job.getDataFilePath());
    StrJobPayload payload = new StrJobPayload(job.getId(), jobData);
    messagePublisher.publishDispatcherPayload(node.getNodeIp(), payload);
    node.setTokens(node.getTokens() - 1);
    nodeDao.update(node);
    job.setStatus(JobStatus.PROCESSING);
    job.setStartProcessing(new Date());
    job.setNode(node);
    jobDao.update(job);
  }

  @Override
  public void dispatchJob(String nodeIp, StrJobPayload payload) throws Exception {
    try {
      sendJobPayload(nodeIp, payload);
    } catch (Exception e) {
      log.error("Error occurred during sending job payload", e);
      log.info("Resending job payload to node {}", nodeIp);
      try {
        sendJobPayload(nodeIp, payload);
      } catch (Exception ex) {
        log.error("Error occurred second time during sending job payload", e);
        errorHandler.handleError(nodeIp, payload, ex);
      }
    }
  }

  private void sendJobPayload(String nodeIp, StrJobPayload payload) throws Exception {
    log.info("Enqueueing payload {} in node {}", payload, nodeIp);
    MessageChannel channel = queueLocator.getNodeJobChannel(nodeIp);
    GenericMessage<StrJobPayload> msg = new GenericMessage<StrJobPayload>(payload);
    log.debug("Sending message");
    channel.send(msg);
    log.debug("Message sent");
  }

  private int sumOfTokens(List<Node> nodes) {
    int sum = 0;

    for (Node node : nodes) {
      sum += node.getTokens();
    }

    return sum;
  }

  @Override
  @Transactional(propagation = Propagation.REQUIRED, isolation = Isolation.SERIALIZABLE)
  public void persistJobEnd(Long jobId, Long nodeId) throws Exception {
    Node node = nodeDao.getByIdForUpdate(nodeId);
    Job job = jobDao.getById(jobId);
    if (!job.getStatus().equals(JobStatus.PROCESSING)) {
      throw new ManagerException("Job with id=" + job.getId() + " has " + job.getStatus()
                                 + " status. Can't persist end of such job.");
    }
    JobGroup jobGroup = jobGroupDao.getByIdForUpdate(job.getJobGroupId());

    job.setStatus(JobStatus.FINISHED);
    job.setEndProcessing(new Date());

    jobDao.update(job);

    jobDataDao.remove(job.getDataFilePath());
    jobGroup.setFinishedJobs(jobGroup.getFinishedJobs() + 1);
    if (jobGroup.getOverallJobs().equals(jobGroup.getFinishedJobs())) {
      jobGroup.setEndDate(new Date());
    }
    jobGroup = jobGroupDao.update(jobGroup);

    node.setTokens(node.getTokens() + 1);
    node = nodeDao.update(node);
  }

  public void setJobDao(JobDao jobDao) {
    this.jobDao = jobDao;
  }

  public void setJobDataDao(JobDataDao jobDataDao) {
    this.jobDataDao = jobDataDao;
  }

  public void setNodeDao(NodeDao nodeDao) {
    this.nodeDao = nodeDao;
  }

  public void setJobGroupDao(JobGroupDao jobGroupDao) {
    this.jobGroupDao = jobGroupDao;
  }

  public void setMessagePublisher(ManagerMessagePublisher messagePublisher) {
    this.messagePublisher = messagePublisher;
  }

  public void setQueueLocator(QueueLocator queueLocator) {
    this.queueLocator = queueLocator;
  }

  public void setErrorHandler(NodeCommunicationErrorHandler errorHandler) {
    this.errorHandler = errorHandler;
  }
}
